package com.hxk.juc.thread.prodCons;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 使用BlockQueue阻塞队列实现生产者消费者案例
 */
public class BlockQueueProdConsCase {

    public static void main(String[] args) {
        BQData bqData = new BQData(new ArrayBlockingQueue<>(3));
        //生产线程
        new Thread(()->{
            System.out.println(Thread.currentThread().getName()+"\t 生产线程启动......");
            bqData.prodM();},"生产者").start();
        //消费线程
        new Thread(()->{
            System.out.println(Thread.currentThread().getName()+"\t 消费线程启动......");
            bqData.consM();},"消费者").start();

        //生产5秒后停止生产
        try {TimeUnit.SECONDS.sleep(5);} catch (InterruptedException e) {e.printStackTrace();}
        System.out.println();
        System.out.println();
        System.out.println();

        bqData.stop();
    }
}

class BQData{

    //生产和消费的阀值
    private volatile boolean flag = true;

    //要生产的产品（保证数据的原子性）
    private AtomicInteger atomicInteger = new AtomicInteger();

    //阻塞队列
    private BlockingQueue<String> blockingQueue = null;

    public BQData(BlockingQueue<String> blockingQueue){
        this.blockingQueue = blockingQueue;
    }

    //生产方法
    public void prodM(){
        //判断是否生产
        String data = null;
        boolean isSeccuss = false;
        while (flag){
            data = atomicInteger.incrementAndGet()+"";
            try {
                isSeccuss = blockingQueue.offer(data, 2L, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            if(isSeccuss){
                System.out.println(Thread.currentThread().getName()+"\t 插入队列"+data+"成功！");
            }else{
                System.out.println(Thread.currentThread().getName()+"\t 插入队列"+data+"失败！");
            }
            try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}
        }
        System.out.println(Thread.currentThread().getName()+"\t 主线程叫停、停止生产");
    }

    public void consM(){
        String rsVal = null;
        while (flag){
            try {
                rsVal = blockingQueue.poll(2L, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            if(null == rsVal || "".equalsIgnoreCase(rsVal)){
               flag = false;
                System.out.println(Thread.currentThread().getName() + "\t超过2秒钟没有消费，退出消费");
                System.out.println();
                System.out.println();
                System.out.println();
                return;
            }
            System.out.println(Thread.currentThread().getName()+"\t 消费队列" + rsVal + "成功！");
        }
    }

    public void stop(){
        this.flag = false;
    }
}
